package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{KafkaUtil, SparkUtil, DateUtil, NoticeMsg}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext, _}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.{HashMap, ArrayBuffer}

/**
 * Created by Lenovo on 2018/8/12.
 * 计算运动:有数据且连续两天运动步数<4000
 */
object CountOneDayStepsStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    println(args.mkString("\t"))

    val Array(brokers, sugarTopicList, topicList, mqHost, mqPort, mqUsername, mqPassword,
    mqVirtualHost, exchangeName, mqRouteKey,checkpointDir, duration, rddDuration,  sendTime, appName
    ) = args
    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[StepsRecord]))

//    val checkpointPath = "/data/streaming/checkpoint/count_oneday_stepstream/"

    val sc = SparkContext.getOrCreate(conf)


    val updateSugarFunc = (values: Seq[Long], state: Option[Long]) => {
      //保存状态
      val previousCount = state.getOrElse(0L)
      val result = if (previousCount == 0) values.min
      else if (values.isEmpty) previousCount
      else math.min(values.min, previousCount)
      Some(result)
    }

    val updateFunc = (values: Seq[HashMap[String, Int]],
                      state: Option[HashMap[String, Int]]) => {
      //保存状态
      val twoDaysAgo = DateUtil.getTodayDateTime("yyyyMMdd") - 2 * 24 * 60 * 60 * 1000
      val currentCount = HashMap[String, Int]()
      values.foreach {
        case map =>
          map.foreach {
            case (day, steps) =>
              if (currentCount.contains(day)) currentCount(day) += steps
              else currentCount(day) = steps
          }
      }
      val previousCount = state.getOrElse(HashMap[String, Int]())
      previousCount.foreach {
        case (day, steps) =>
          val recordDay = DateUtil.getDateTimestamp(day)
          if (recordDay >= twoDaysAgo) {
            if (currentCount.contains(day)) currentCount(day) += steps
            else currentCount(day) = steps
          }
      }
      Some(currentCount)
    }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duration.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(
        mqHost,
        mqPort.toInt,
        mqUsername,
        mqPassword,
        mqVirtualHost
      )

      val topics = topicList.split(",").toSet
      val sugarTopics = sugarTopicList.split(",").toSet

      val stepMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          topics
        )

      val sugarMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams, sugarTopics )
      val sugarBatch: DStream[(String, Long)] =
        sugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, vo.timestamp)
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", 0L))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateSugarFunc)

      sugarBatch.checkpoint(Duration(rddDuration.toInt  * 1000))

      //@todo
      val stepsBatch: DStream[(String, HashMap[String, Int])] =
        stepMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList = gson.fromJson(msg._2, classOf[Array[StepsRecord]]).map { vo => {
                val daySteps = HashMap[String, Int]()
                daySteps += ((DateUtil.getTimeStampFormat(vo.timestamp), vo.steps))
                (vo.personId.toString, daySteps)
              }
              }
              voList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", HashMap[String, Int]()))
            }
          }.filter(_._1.nonEmpty)
          .updateStateByKey(updateFunc)

      stepsBatch.checkpoint(Duration(rddDuration.toInt * 1000))

      val sugarStep = sugarBatch.leftOuterJoin(stepsBatch)

      sugarStep.foreachRDD(mess =>
        if (!mess.isEmpty()) {
          mess.foreachPartition(lines => {
            val today = DateUtil.getTodayFormat()
            val todayDate = DateUtil.getTodayDateTime("yyyyMMdd")
            val nowStamp = DateUtil.getTodayDateTime()
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, (sugarTime, stepMap)) => {
                println("____ pid :" + pid + ",,, sugarTime:" + sugarTime + ",,, sugarTime:" + stepMap)
                // @todo excel 10: 有数据且连续两天运动步数<4000
                if (nowStamp - todayDate >= sendTime.toInt * 60 * 60 * 1000 && nowStamp - todayDate < sendTime.toInt * 60 * 60 * 1000 + duration.toInt * 1000 &&
                  ((stepMap.isEmpty && nowStamp - sugarTime > 2 * 24 * 60 * 60 * 1000) ||
                    (stepMap.nonEmpty && stepMap.get.filter(_._1 != today).values.sum < 4000))
                ) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.low_step._1,
                    NoticeMsg.low_step._2,
                    NoticeMsg.low_step._3,
                    nowStamp.toString,
                    NoticeMsg.low_step._5,
                    NoticeMsg.low_step._4,
                    NoticeMsg.low_step._6
                  )
                  mqMsg += gson.toJson(msg)
                }
              }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          })

        })

      ssc
    }


    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}